package com.sdses.flink.udf;

import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.google.common.io.BaseEncoding;
import io.debezium.util.HexConverter;
import org.apache.flink.table.functions.FunctionContext;
import org.apache.flink.table.functions.ScalarFunction;
import org.locationtech.jts.geom.Geometry;
import org.locationtech.jts.io.WKBReader;

import java.util.Base64;

public class PgGeometryTypeConvert extends ScalarFunction {

    private ObjectMapper objectMapper = null;
    private WKBReader reader = null;

    @Override
    public void open(FunctionContext context) throws Exception {
        reader = new WKBReader();
        objectMapper = new ObjectMapper();
    }

    public String eval(String json) throws Exception {
        JsonNode jsonNode = objectMapper.readTree(json);
        String hexewkb = jsonNode.get("hexewkb").asText();
        int srid = jsonNode.get("srid").asInt();
        byte[] wkbBytes = BaseEncoding.base16().lowerCase().decode(hexewkb);

//        byte[] wkbBytes = HexConverter.convertFromHex(wkb);
        Geometry geom = reader.read(wkbBytes);
        geom.setSRID(srid);
        return geom.toString() ;
    }

}
